#include "app_pool.h"

#include "pthread.h"
#include "mqueue.h"
#include "log/log.h"
#include <stdlib.h>
#include <stdio.h>

/* 全局变量  */
pthread_t *thread_ids = NULL; /* 存储所有的线程标识符 */
int thread_nums = 0;

mqd_t mq_id = 0;
char *mq_name = "/app_pool";

/**
 * @function 线程回调函数，线程池里面的线程们通过消息队列接收回调函数执行
 *           谁抢到谁执行，没抢到就阻塞
 */
void *thread_callback(void *parm)
{
    Task thread_task;
    while (1)
    {
        /* 1.获取任务
          队列为空就阻塞
          return -1: 获取失败
          return >=0:表示获取的实际字节数
        */
        if (mq_receive(mq_id, (char *)&thread_task, sizeof(thread_task), NULL) >= 0)
        {
            /* 2.执行任务 */
            thread_task.taskCallBack(thread_task.parm);
        }
    }
}

/**
 * @function 创建线程池
 */
returnType_t app_pool_init(int thread_size)
{
    /* 1. create threads */
    thread_nums = thread_size;
    thread_ids = (pthread_t *)malloc(sizeof(pthread_t) * thread_nums); /* 存储所有的线程标识符 */
    if (thread_ids == NULL)
    {
        log_error("malloc thread_ids failed");
        return ERROR;
    }

    for (int i = 0; i < thread_nums; i++)
    {
        /*
         parm1: thread_id       parm2:thread_attribute
         parm3:thread_callback  parm4:thread_callback_parm
        */
        pthread_create(&thread_ids[i], NULL, thread_callback, NULL);
    }

    return SUCCESS;
}

/**
 *@function 注册线程执行的回调函数，说白了就是将要执行的任务塞到消息队列里面
 */
returnType_t app_pool_register(void (*callback)(void *), void *parm)
{

    /* 1. 创建/打开队列  */
    struct mq_attr attr =
        {
            .mq_maxmsg = 10,
            .mq_msgsize = sizeof(Task) /* 往队列里面塞的是啥？是回调函数*/
        };

    mq_id = mq_open(mq_name, O_CREAT | O_RDWR, 0755, &attr);
    if (mq_id == -1)
    {
        log_error("mq_open failed");
        // exit(1);/* 调用这可以直接清理缓存*/
        return ERROR;
    }

    /* 2. 往消息队列里面塞任务 */
    Task thread_task =
        {
            .taskCallBack = callback,
            .parm = parm};

    if (mq_send(mq_id, (char *)&thread_task, sizeof(thread_task), 0) == -1)
    {
        log_error("mq_send failed");
        return ERROR;
    }

    /* 3.关闭队列 */
    if (mq_close(mq_id) == -1)
    {
        log_error("mq_close failed");

        return ERROR;
    }

    return SUCCESS;
}

/**
 * @function 销毁线程池
 */
returnType_t app_pool_destroy(void)
{
    /* 销毁线程池 */
    for (int i = 0; i < thread_nums; i++)
    {
        pthread_cancel(thread_ids[i]);     /* 杀死对应ID的标识符线程 */
        pthread_join(thread_ids[i], NULL); /* 阻塞式等待对应线程结束，并收尸 */
    }

    /* 释放堆区内存*/
    free(thread_ids);

    /* 销毁消息队列*/
    mq_unlink(mq_name);

    return SUCCESS;
}